package com.wuwangfu.state;

import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.queryablestate.client.QueryableStateClient;

import java.util.concurrent.CompletableFuture;

/**
 * @Description：查询状态客户端
 * @Author：jcshen
 * @Date：2023-06-28
 *
 */
public class QueryableKeyedStateClient {

    public static void main(String[] args) throws Exception {
        QueryableStateClient client = new QueryableStateClient("localhost", 9069);
        //初始化状态数据或恢复历史状态数据
        ValueStateDescriptor<Integer> stateDestor = new ValueStateDescriptor<>(
                "word-state",//指定状态描述器名称
                Integer.class//存储数据的类型
        );
        //根据条件查询状态
        CompletableFuture<ValueState<Integer>> resultFuture = client.getKvState(
                JobID.fromHexString("5bbaf46b7ba795bca67d3af44f26546e"),//job的ID
                "query-word-count",//可查询的状态名称
                "flink",//查询的key
                BasicTypeInfo.STRING_TYPE_INFO,
                stateDestor
        );
        //获取状态结果
        resultFuture.thenAccept(response->{
            try {
                Integer res = response.value();
                System.out.println(res);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });
        Thread.sleep(5000);

    }
}
